package org.apache.flink.runtime.resourceconsumer;

import static org.apache.flink.util.Preconditions.checkNotNull;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeoutException;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import org.apache.elasticresource.api.BusinessExecution;
import org.apache.elasticresource.api.BusinessResourceConsumer;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ConsumingStatus;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.ClusterOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.runtime.allocatedeploy.DefaultExecutionOperations;
import org.apache.flink.runtime.allocatedeploy.ExecutionDeployer;
import org.apache.flink.runtime.allocatedeploy.ExecutionSlotAllocator;
import org.apache.flink.runtime.allocatedeploy.ExecutionSlotAllocatorFactory;
import org.apache.flink.runtime.allocatedeploy.FixExecutionSlotAllocationContext;
import org.apache.flink.runtime.allocatedeploy.SimpleExecutionSlotAllocator;
import org.apache.flink.runtime.allocatedeploy.execution.Execution;
import org.apache.flink.runtime.allocatedeploy.execution.ExecutionID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.consumermanager.OnCompletionActions;
import org.apache.flink.runtime.heartbeat.HeartbeatListener;
import org.apache.flink.runtime.heartbeat.HeartbeatManager;
import org.apache.flink.runtime.heartbeat.HeartbeatReceiver;
import org.apache.flink.runtime.heartbeat.HeartbeatSender;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.heartbeat.NoOpHeartbeatManager;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.metrics.groups.ConsumerManagerConsumerMetricGroup;
import org.apache.flink.runtime.providermanager.ProviderManagerLocation;
import org.apache.flink.runtime.providermanager.ProviderManagerLocation.ResolutionMode;
import org.apache.flink.runtime.providermanager.ResourceAllocationState;
import org.apache.flink.runtime.providermanager.UnresolvedProviderManagerLocation;
import org.apache.flink.runtime.registration.RegisteredRpcConnection;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.registration.RetryingRegistration;
import org.apache.flink.runtime.resourceconsumer.factories.ConsumerManagerConsumerMetricGroupFactory;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
import org.apache.flink.runtime.resourceprovider.ResourceProviderGateway;
import org.apache.flink.runtime.resourceprovider.ResourceProviderToResourceConsumerHeartbeatPayload;
import org.apache.flink.runtime.resourceprovider.slot.SlotOffer;
import org.apache.flink.runtime.rest.messages.ThreadDumpInfo;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.FencedRpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.RpcServiceUtils;
import org.apache.flink.runtime.slotpool.DeclarativeSlotPoolFactory;
import org.apache.flink.runtime.slotpool.DefaultDeclarativeSlotPoolFactory;
import org.apache.flink.runtime.slotpool.PhysicalSlotProvider;
import org.apache.flink.runtime.slotpool.PhysicalSlotProviderImpl;
import org.apache.flink.runtime.slotpool.SlotPool;
import org.apache.flink.runtime.slotpool.SlotPoolService;
import org.apache.flink.runtime.slotpool.SlotPoolServiceFactory;
import org.apache.flink.runtime.slotpool.SlotSelectionStrategy;
import org.apache.flink.runtime.slots.ResourceRequirement;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.runtime.util.SlotSelectionStrategyUtils;
import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.apache.flink.util.ERException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.concurrent.FutureUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @类名: ResourceConsumer
 * @说明: 资源消费者
 *        > 分配资源
 *        > 申请资源
 *        > 资源slot池
 *        ID：JobID(consumerId), ConsumerMasterId, ResourceId
 *        用户继承，实现消费行为
 * @author   leehom
 * @Date	 2023年10月17日 下午1:47:40
 * 修改记录：
 *
 * @see 	 
 */
public class ResourceConsumer extends FencedRpcEndpoint<ConsumingMasterId>
        						implements ResourceConsumerGateway, ResourceConsumerService {

    /** names for 弹性资源 distributed components. */
    public static final String RESOURCE_CONSUMER_NAME = "resourceconsumer";

    /** The log object used for debugging. */
    static final Logger LOG = LoggerFactory.getLogger(ResourceConsumer.class);
    
    // ------------------------------------------------------------------------
    private final ResourceConsumerConfiguration resourceConsumerConfiguration;
    
    /** 消费者实例Id，一个资源消费者同时可有多个消费者实例*/
    private final ConsumingMasterId consumingMasterId;
    /** */
    private final String consumerName;
    
    /** 组件Id*/
    private final ResourceID resourceId;
    
    /** 消费者Id*/
    private final ConsumerID consuemrId;

    private final HighAvailabilityServices highAvailabilityServices;

    private final HeartbeatServices heartbeatServices;
    
    private final ScheduledExecutorService futureExecutor;
    
    private final Executor ioExecutor;
    
    private final ConsumerManagerConsumerMetricGroup consumerManagerConsumerMetricGroup;

    /** */
    private final OnCompletionActions resumeCompletionActions;

    private final FatalErrorHandler fatalErrorHandler;

    /** 资源slot池*/
    private final SlotPoolService slotPoolService;

    private final boolean retrieveTaskManagerHostName;

    // --------- ResourceManager --------
    private final LeaderRetrievalService resourceManagerLeaderRetriever;

    // --------- 资源提供者 --------
    private final Map<ResourceID, ResourceProviderRegistration> registeredResourceProviders;
    
    // --------- 业务资源提供者----------
    private final BusinessResourceConsumer businessResourceConsumer;
    // 
    protected final ExecutionSlotAllocator executionSlotAllocator;
    protected final ExecutionDeployer executionDeployer;

    // -------- Mutable(易边的) fields ---------
    @Nullable private ResourceManagerAddress resourceManagerAddress;
    @Nullable private ResourceManagerConnection resourceManagerConnection;
    @Nullable private EstablishedResourceManagerConnection establishedResourceManagerConnection;
    /** 任务管理器心跳，带载荷*/
    private HeartbeatManager<ResourceProviderToResourceConsumerHeartbeatPayload, AllocatedSlotReport> resourceConsumerHeartbeatManager;
    /** 资源管理器心跳管理*/
    private HeartbeatManager<Void, Void> resourceManagerHeartbeatManager;
    
    private final ExecutionDeploymentTracker executionDeploymentTracker;
    private final ExecutionDeploymentReconciler executionDeploymentReconciler;
    
    /** Current status of the job execution. */
    private volatile ConsumingStatus state = ConsumingStatus.CREATED;

    // ------------------------------------------------------------------------
    public ResourceConsumer(
    		ConsumingMasterId consumingMasterId,
    		String consumerName,
            RpcService rpcService,
            ResourceConsumerConfiguration resourceConsumerConfiguration,
            ResourceID resourceId,
            ConsumerID consuemrId,
            HighAvailabilityServices highAvailabilityService,
            SlotPoolServiceFactory slotPoolServiceFactory,
            HeartbeatServices heartbeatServices,
            ConsumerManagerConsumerMetricGroupFactory consumerMetricGroupFactory,
            OnCompletionActions jobCompletionActions,
            ExecutorService ioExecutor,
            ScheduledExecutorService futureExecutor,
            FatalErrorHandler fatalErrorHandler,
            ExecutionDeployer.Factory executionDeployerFactory,
            BusinessResourceConsumer businessResourceConsumer,
            ExecutionDeploymentTracker executionDeploymentTracker,
            ExecutionDeploymentReconciler.Factory executionDeploymentReconcilerFactory
            ) throws Exception {
        super(rpcService, RpcServiceUtils.createRandomName(RESOURCE_CONSUMER_NAME), consumingMasterId);
        //
        final ExecutionDeploymentReconciliationHandler executionStateReconciliationHandler =
                new ExecutionDeploymentReconciliationHandler() {

                    @Override
                    public void onMissingDeploymentsOf(
                            Collection<ExecutionID> executionAttemptIds, ResourceID host) {
                        log.debug(
                                "Failing deployments {} due to no longer being deployed.",
                                executionAttemptIds);

                    }

                    @Override
                    public void onUnknownDeploymentsOf(
                            Collection<ExecutionID> executionAttemptIds, ResourceID host) {
                        log.debug(
                                "Canceling left-over deployments {} on task executor {}.",
                                executionAttemptIds,
                                host);

                    }
                };
                
        this.executionDeploymentTracker = executionDeploymentTracker;
        this.executionDeploymentReconciler =
                executionDeploymentReconcilerFactory.create(executionStateReconciliationHandler);
        //
        this.resourceConsumerConfiguration = checkNotNull(resourceConsumerConfiguration);
        this.resourceId = checkNotNull(resourceId);
		this.consuemrId = consuemrId;
        this.highAvailabilityServices = checkNotNull(highAvailabilityService);
		this.futureExecutor = futureExecutor;
        this.resumeCompletionActions = checkNotNull(jobCompletionActions);
        this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
        this.retrieveTaskManagerHostName =
                resourceConsumerConfiguration
                        .getConfiguration()
                        .getBoolean(ConsumerManagerOptions.RETRIEVE_TASK_MANAGER_HOSTNAME);
        this.consumingMasterId = consumingMasterId;
        this.consumerName = consumerName;
        log.info("Initializing consumer '{}' ({}).", consumerName, consumingMasterId);
        // 资源管理器leader获取服务
        resourceManagerLeaderRetriever =
                highAvailabilityServices.getResourceManagerLeaderRetriever();
        //
        this.registeredResourceProviders = new HashMap<>();
        //
        this.slotPoolService =
                checkNotNull(slotPoolServiceFactory)
                        .createSlotPoolService(
                        		consuemrId,
                                createDeclarativeSlotPoolFactory(
                                        resourceConsumerConfiguration.getConfiguration()));
        final SlotPool slotPool =
                slotPoolService
                        .castInto(SlotPool.class)
                        .orElseThrow(
                                () -> new IllegalStateException(
                                                "The ResourceConsumer requires a SlotPool."));
        final ExecutionSlotAllocatorFactory allocatorFactory =
                createExecutionSlotAllocatorFactory(resourceConsumerConfiguration.getConfiguration(), slotPool);
        
        // metrics分组
        this.consumerManagerConsumerMetricGroup = consumerMetricGroupFactory.create(consuemrId, consumerName);
        // 服务
        this.heartbeatServices = checkNotNull(heartbeatServices);
        this.resourceConsumerHeartbeatManager = NoOpHeartbeatManager.getInstance();
        this.resourceManagerHeartbeatManager = NoOpHeartbeatManager.getInstance();
        // 
        this.resourceManagerConnection = null;
        this.establishedResourceManagerConnection = null;
        //
        this.businessResourceConsumer = businessResourceConsumer;
        // 资源分配器
        ResourceProfile resourceProfile = ResourceProfile.fromResources(0, 0);
        this.executionSlotAllocator =
                checkNotNull(allocatorFactory)
                        .createInstance(new FixExecutionSlotAllocationContext(resourceProfile));
        // 部署器
        this.executionDeployer =
                executionDeployerFactory.createInstance(
                        log,
                        executionSlotAllocator,
                        new DefaultExecutionOperations(),
                        (ExecutionID, AllocationID)->{},
                        this.getMainThreadExecutor());
        // 构建io Executor
        final int numberCPUCores = Hardware.getNumberCPUCores();
        Configuration config = resourceConsumerConfiguration.getConfiguration();
        final int consumerManagerIoPoolSize =
                config.getInteger(ConsumerManagerOptions.CONSUMER_MANAGER_IO_POOL_SIZE, numberCPUCores);
        this.ioExecutor =
                Executors.newFixedThreadPool(
                        consumerManagerIoPoolSize, new ExecutorThreadFactory("consumermanager-io"));
    }

    // 心跳
    private HeartbeatManager<Void, Void> createResourceManagerHeartbeatManager(
            HeartbeatServices heartbeatServices) {
        return heartbeatServices.createHeartbeatManager(
                resourceId, new ResourceManagerHeartbeatListener(), getMainThreadExecutor(), log);
    }

    // 
    private HeartbeatManager<ResourceProviderToResourceConsumerHeartbeatPayload, AllocatedSlotReport>
            createTaskManagerHeartbeatManager(HeartbeatServices heartbeatServices) {
        return heartbeatServices.createHeartbeatManagerSender(
                resourceId, new ResourceProviderHeartbeatListener(), getMainThreadExecutor(), log);
    }

    private DeclarativeSlotPoolFactory createDeclarativeSlotPoolFactory(
            Configuration configuration) {
    	return new DefaultDeclarativeSlotPoolFactory();
    }

    // 分配和部署
    public void allocateSlotsAndDeploy(
            final List<Execution> executionsToDeploy) {
    	this.executionDeployer.allocateSlotsAndDeploy(executionsToDeploy);
    }
    // ----------------------------------------------------------------------------------------------
    // rpc Lifecycle management
    // ----------------------------------------------------------------------------------------------
    @Override
    protected void onStart() throws ERException {
        try {
            startResourceConsuming();
        } catch (Exception e) {
            final ResoruceConsumerException jobMasterException =
                    new ResoruceConsumerException("Could not start the ResourceConsumer.", e);
            handleJobMasterError(jobMasterException);
            throw jobMasterException;
        }
    }

    /** Suspend the consuming and shutdown all other services including rpc. */
    @Override
    public CompletableFuture<Void> onStop() {
        log.info(
                "Stopping the ResourceConsumer '{}' ({}).",
                consumerName, consuemrId);

        // make sure there is a graceful exit
        return stopResourceConsuming(
                        new ERException(
                                String.format(
                                        "Stopping ResourceConsumer for job '%s' (%s).",
                                        consumerName, consuemrId)))
                .exceptionally(
                        exception -> {
                            throw new CompletionException(
                                    new ResoruceConsumerException(
                                            "Could not properly stop the JobMaster.", exception));
                        });
    }

    // ----------------------------------------------------------------------------------------------
    // RPC methods
    // ----------------------------------------------------------------------------------------------

    @Override
    public CompletableFuture<Acknowledge> cancel(Time timeout) {
        return CompletableFuture.completedFuture(Acknowledge.get());
    }

    /**
     * 更新任务执行状态，资源消费用户实现
     * Updates the task execution state for a given task.
     *
     * @param taskExecutionState New task execution state for a given task
     * @return Acknowledge the task execution state update
     */
    @Override
    public CompletableFuture<Acknowledge> updateTaskExecutionState(
            final ResourceAllocationState taskExecutionState) {
    	// 。。。
    	return new CompletableFuture<Acknowledge>();
    }

    @Override
    public CompletableFuture<Acknowledge> disconnectTaskManager(
            final ResourceID resourceID, final Exception cause) {
        log.info(
                "Disconnect ResourceProvider {} because: {}",
                resourceID.getStringWithMetadata(),
                cause.getMessage(),
                ExceptionUtils.returnExceptionIfUnexpected(cause.getCause()));
        ExceptionUtils.logExceptionIfExcepted(cause.getCause(), log);

        resourceConsumerHeartbeatManager.unmonitorTarget(resourceID);
        slotPoolService.releaseTaskManager(resourceID, cause);

        ResourceProviderRegistration taskManagerRegistration = registeredResourceProviders.remove(resourceID);

        if (taskManagerRegistration != null) {
            taskManagerRegistration
                    .getTaskExecutorGateway()
                    .disconnectResourceConsumer(consuemrId, cause);

        }

        return CompletableFuture.completedFuture(Acknowledge.get());
    }

    // 接收资源提供者提供的资源slot
    @Override
    public CompletableFuture<Collection<SlotOffer>> offerSlots(
            final ResourceID taskManagerId, final Collection<SlotOffer> slots, final Time timeout) {
    	// 获取注册的任务管理器
        ResourceProviderRegistration resourceProviderRegistration = registeredResourceProviders.get(taskManagerId);
        if (resourceProviderRegistration == null) {
            return FutureUtils.completedExceptionally(
                    new Exception("Unknown TaskManager " + taskManagerId));
        }
        //
        final RpcProviderManagerGateway rpcTaskManagerGateway =
                new RpcProviderManagerGateway(
                        resourceProviderRegistration.getTaskExecutorGateway(), getFencingToken());

        return CompletableFuture.completedFuture(
                slotPoolService.offerSlots(
                        resourceProviderRegistration.getTaskManagerLocation(),
                        rpcTaskManagerGateway,
                        slots));
    }

    @Override
    public void failSlot(
            final ResourceID taskManagerId,
            final AllocationID allocationId,
            final Exception cause) {

        if (registeredResourceProviders.containsKey(taskManagerId)) {
            internalFailAllocation(taskManagerId, allocationId, cause);
        } else {
            log.warn(
                    "Cannot fail slot "
                            + allocationId
                            + " because the TaskManager "
                            + taskManagerId
                            + " is unknown.");
        }
    }

    private void internalFailAllocation(
            @Nullable ResourceID resourceId, AllocationID allocationId, Exception cause) {
        final Optional<ResourceID> resourceIdOptional =
                slotPoolService.failAllocation(resourceId, allocationId, cause);
        resourceIdOptional.ifPresent(
                taskManagerId -> {
                	releaseEmptyTaskManager(taskManagerId);
                });
    }

    private void releaseEmptyTaskManager(ResourceID resourceId) {
        disconnectTaskManager(
                resourceId,
                new ERException(
                        String.format(
                                "No more slots registered at JobMaster %s.",
                                resourceId.getStringWithMetadata())));
    }

    @Override
    public CompletableFuture<RegistrationResponse> registerResourceProvider(
            final ConsumerID consumerId,
            final ResourceProviderRegistrationInformation taskManagerRegistrationInformation,
            final Time timeout) {

        if (!this.consuemrId.equals(consumerId)) {
            log.debug(
                    "Rejecting ResourceProvider registration attempt because of wrong consumer id {}.",
                    consumerId);
            return CompletableFuture.completedFuture(
                    new RCRPRegistrationRejection(
                            String.format(
                                    "The ResourceConsumer is not responsible for consumer %s. Maybe the TaskManager used outdated connection information.",
                                    consumerId)));
        }

        final ProviderManagerLocation providerManagerLocation;
        try {
            providerManagerLocation =
                    resolveProviderManagerLocation(
                            taskManagerRegistrationInformation.getUnresolvedTaskManagerLocation());
        } catch (ERException exception) {
            log.error("Could not accept TaskManager registration.", exception);
            return CompletableFuture.completedFuture(new RegistrationResponse.Failure(exception));
        }

        final ResourceID taskManagerId = providerManagerLocation.getResourceID();
        final UUID sessionId = taskManagerRegistrationInformation.getTaskManagerSession();
        final ResourceProviderRegistration taskManagerRegistration =
                registeredResourceProviders.get(taskManagerId);

        if (taskManagerRegistration != null) {
            if (taskManagerRegistration.getSessionId().equals(sessionId)) {
                log.debug(
                        "Ignoring registration attempt of TaskManager {} with the same session id {}.",
                        taskManagerId,
                        sessionId);
                final RegistrationResponse response = new RCRPRegistrationSuccess(resourceId);
                return CompletableFuture.completedFuture(response);
            } else {
                disconnectTaskManager(
                        taskManagerId,
                        new ERException(
                                String.format(
                                        "A registered TaskManager %s re-registered with a new session id. This indicates a restart of the TaskManager. Closing the old connection.",
                                        taskManagerId)));
            }
        }

        return getRpcService()
                .connect(
                        taskManagerRegistrationInformation.getTaskManagerRpcAddress(),
                        ResourceProviderGateway.class)
                .handleAsync(
                        (ResourceProviderGateway taskExecutorGateway, Throwable throwable) -> {
                            if (throwable != null) {
                                return new RegistrationResponse.Failure(throwable);
                            }

                            slotPoolService.registerTaskManager(taskManagerId);
                            registeredResourceProviders.put(
                                    taskManagerId,
                                    ResourceProviderRegistration.create(
                                            providerManagerLocation, taskExecutorGateway, sessionId));

                            // monitor the task manager as heartbeat target
                            resourceConsumerHeartbeatManager.monitorTarget(
                                    taskManagerId,
                                    new ResourceProviderHeartbeatSender(taskExecutorGateway));

                            return new RCRPRegistrationSuccess(resourceId);
                        },
                        getMainThreadExecutor());
    }

    @Nonnull
    private ProviderManagerLocation resolveProviderManagerLocation(
            UnresolvedProviderManagerLocation unresolvedTaskManagerLocation) throws ERException {
        try {
            if (retrieveTaskManagerHostName) {
                return ProviderManagerLocation.fromUnresolvedLocation(
                        unresolvedTaskManagerLocation, ResolutionMode.RETRIEVE_HOST_NAME);
            } else {
                return ProviderManagerLocation.fromUnresolvedLocation(
                        unresolvedTaskManagerLocation, ResolutionMode.USE_IP_ONLY);
            }
        } catch (Throwable throwable) {
            final String errMsg =
                    String.format(
                            "Provider Manager address %s cannot be resolved. %s",
                            unresolvedTaskManagerLocation.getExternalAddress(),
                            throwable.getMessage());
            throw new ERException(errMsg, throwable);
        }
    }

    @Override
    public void disconnectResourceManager(
            final ResourceManagerId resourceManagerId, final Exception cause) {

        if (isConnectingToResourceManager(resourceManagerId)) {
            reconnectToResourceManager(cause);
        }
    }

    private boolean isConnectingToResourceManager(ResourceManagerId resourceManagerId) {
        return resourceManagerAddress != null
                && resourceManagerAddress.getResourceManagerId().equals(resourceManagerId);
    }

    @Override
    public CompletableFuture<Void> heartbeatFromTaskManager(
            final ResourceID resourceID, ResourceProviderToResourceConsumerHeartbeatPayload payload) {
        return resourceConsumerHeartbeatManager.receiveHeartbeat(resourceID, payload);
    }

    @Override
    public CompletableFuture<Void> heartbeatFromResourceManager(final ResourceID resourceID) {
        return resourceManagerHeartbeatManager.requestHeartbeat(resourceID, null);
    }

    @Override
    public void notifyNotEnoughResourcesAvailable(
            Collection<ResourceRequirement> acquiredResources) {
        slotPoolService.notifyNotEnoughResourcesAvailable(acquiredResources);
    }

    // ----------------------------------------------------------------------------------------------
    // Internal methods
    // ----------------------------------------------------------------------------------------------

    // -- consume starting and stopping
    // -----------------------------------------------------------------

    private void startResourceConsuming() throws Exception {
        validateRunsInMainThread();

        startResourceConsumerServices();
        //
        this.state = ConsumingStatus.RUNNING;
        
        log.info(
                "Starting consuming of consumer '{}' ({}) under master id {}.",
                this.consumerName,
                this.consuemrId,
                getFencingToken());
        //
        startAllocateAndDeploy();
    }

    //
    private void startAllocateAndDeploy() {
        List<BusinessExecution> bes = businessResourceConsumer.getExecution();
        // 转换为Execution
        List<Execution> executionsToDeploy = Lists.newArrayList();
        for(BusinessExecution be : bes) {
        	Execution e = new Execution(
        			this.consuemrId,
        			be.getCommandKey(),
            		be.getCommand(),
            		be.getSerArgs(),
            		be.getResourceKey(),
            		ioExecutor,
            		System.currentTimeMillis(),
                    resourceConsumerConfiguration.getRpcTimeout(),
                    this.getMainThreadExecutor());
        }
        //
        this.executionDeployer.allocateSlotsAndDeploy(executionsToDeploy);
    }
    
    private void startResourceConsumerServices() throws Exception {
        try {
            this.resourceConsumerHeartbeatManager = createTaskManagerHeartbeatManager(heartbeatServices);
            this.resourceManagerHeartbeatManager =
                    createResourceManagerHeartbeatManager(heartbeatServices);

            // start the slot pool make sure the slot pool now accepts messages for this leader
            slotPoolService.start(getFencingToken(), getAddress(), getMainThreadExecutor());

            // consumer is ready to go, try to establish connection with resource manager
            //   - activate leader retrieval for the resource manager
            //   - on notification of the leader, the connection will be established and
            //     the slot pool will start requesting slots
            resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
        } catch (Exception e) {
            handleStartJobMasterServicesError(e);
        }
    }

    private void handleStartJobMasterServicesError(Exception e) throws Exception {
        try {
            stopResourceConsumerServices();
        } catch (Exception inner) {
            e.addSuppressed(inner);
        }

        throw e;
    }

    private void stopResourceConsumerServices() throws Exception {
        Exception resultingException = null;
        try {
            resourceManagerLeaderRetriever.stop();
        } catch (Exception e) {
            resultingException = e;
        }

        // TODO: Distinguish between job termination which should free all slots and a loss of
        // leadership which should keep the slots
        slotPoolService.close();

        stopHeartbeatServices();

        ExceptionUtils.tryRethrowException(resultingException);
    }

    //
    private CompletableFuture<Void> stopResourceConsuming(final Exception cause) {
        validateRunsInMainThread();        
        CompletableFuture<Void> f1 = CompletableFuture.runAsync(
                () -> { disconnectTaskManagerResourceManagerConnections(cause); 
                		}
        		);
        return FutureUtils.runAfterwards(
                f1,
                () -> {
                	stopResourceConsumerServices();
                });
        
    }

    private void disconnectTaskManagerResourceManagerConnections(Exception cause) {
        // disconnect from all registered TaskExecutors
        final Set<ResourceID> taskManagerResourceIds =
                new HashSet<>(registeredResourceProviders.keySet());

        for (ResourceID taskManagerResourceId : taskManagerResourceIds) {
            disconnectTaskManager(taskManagerResourceId, cause);
        }

        // disconnect from resource manager:
        closeResourceManagerConnection(cause);
    }

    private void stopHeartbeatServices() {
        resourceConsumerHeartbeatManager.stop();
        resourceManagerHeartbeatManager.stop();
    }

    // ----------------------------------------------------------------------------------------------

    private void handleJobMasterError(final Throwable cause) {
        if (ExceptionUtils.isJvmFatalError(cause)) {
            log.error("Fatal error occurred on JobManager.", cause);
            // The fatal error handler implementation should make sure that this call is
            // non-blocking
            fatalErrorHandler.onFatalError(cause);
        } else {
            resumeCompletionActions.resourceConsumerFailed(cause);
        }
    }

    private void notifyOfNewResourceManagerLeader(
            final String newResourceManagerAddress, final ResourceManagerId resourceManagerId) {
        resourceManagerAddress =
                createResourceManagerAddress(newResourceManagerAddress, resourceManagerId);
        //
        reconnectToResourceManager(
                new ERException(
                        String.format(
                                "ResourceManager leader changed to new address %s",
                                resourceManagerAddress)));
    }

    @Nullable
    private ResourceManagerAddress createResourceManagerAddress(
            @Nullable String newResourceManagerAddress,
            @Nullable ResourceManagerId resourceManagerId) {
        if (newResourceManagerAddress != null) {
            // the contract is: address == null <=> id == null
            checkNotNull(resourceManagerId);
            return new ResourceManagerAddress(newResourceManagerAddress, resourceManagerId);
        } else {
            return null;
        }
    }

    private void reconnectToResourceManager(Exception cause) {
        closeResourceManagerConnection(cause);
        tryConnectToResourceManager();
    }

    private void tryConnectToResourceManager() {
        if (resourceManagerAddress != null) {
            connectToResourceManager();
        }
    }

    private void connectToResourceManager() {
        assert (resourceManagerAddress != null);
        assert (resourceManagerConnection == null);
        assert (establishedResourceManagerConnection == null);

        log.info("Connecting to ResourceManager {}", resourceManagerAddress);

        resourceManagerConnection =
                new ResourceManagerConnection(
                        log,
                        consuemrId,
                        resourceId,
                        getAddress(),
                        getFencingToken(),
                        resourceManagerAddress.getAddress(),
                        resourceManagerAddress.getResourceManagerId(),
                        futureExecutor);

        resourceManagerConnection.start();
    }

    //
    private void establishResourceManagerConnection(final ResourceConsumerRegistrationSuccess success) {
        final ResourceManagerId resourceManagerId = success.getResourceManagerId();

        // verify the response with current connection
        if (resourceManagerConnection != null
                && Objects.equals(
                        resourceManagerConnection.getTargetLeaderId(), resourceManagerId)) {
            log.info(
                    "Consumer successfully registered at ResourceManager, leader id: {}.",
                    resourceManagerId);
            //
            final ResourceManagerGateway resourceManagerGateway =
                    						resourceManagerConnection.getTargetGateway();
            //
            final ResourceID resourceManagerResourceId = success.getResourceManagerResourceId();

            establishedResourceManagerConnection =
                    new EstablishedResourceManagerConnection(
                            resourceManagerGateway, resourceManagerResourceId);
            //
            slotPoolService.connectToResourceManager(resourceManagerGateway);

            resourceManagerHeartbeatManager.monitorTarget(
                    resourceManagerResourceId,
                    new ResourceManagerHeartbeatReceiver(resourceManagerGateway));
        } else {
            log.debug(
                    "Ignoring resource manager connection to {} because it's duplicated or outdated.",
                    resourceManagerId);
        }
    }

    private void closeResourceManagerConnection(Exception cause) {
        if (establishedResourceManagerConnection != null) {
            dissolveResourceManagerConnection(establishedResourceManagerConnection, cause);
            establishedResourceManagerConnection = null;
        }

        if (resourceManagerConnection != null) {
            // stop a potentially ongoing registration process
            resourceManagerConnection.close();
            resourceManagerConnection = null;
        }
    }

    private void dissolveResourceManagerConnection(
            EstablishedResourceManagerConnection establishedResourceManagerConnection,
            Exception cause) {
        final ResourceID resourceManagerResourceID =
                establishedResourceManagerConnection.getResourceManagerResourceID();

        if (log.isDebugEnabled()) {
            log.debug(
                    "Close ResourceManager connection {}.",
                    resourceManagerResourceID.getStringWithMetadata(),
                    cause);
        } else {
            log.info(
                    "Close ResourceManager connection {}: {}",
                    resourceManagerResourceID.getStringWithMetadata(),
                    cause.getMessage());
        }
        //
        resourceManagerHeartbeatManager.unmonitorTarget(resourceManagerResourceID);
        //
        ResourceManagerGateway resourceManagerGateway =
                establishedResourceManagerConnection.getResourceManagerGateway();
        resourceManagerGateway.disconnectConsumer(this.consuemrId, cause);
        slotPoolService.disconnectResourceManager();
    }

    // ----------------------------------------------------------------------------------------------
    // Service methods
    // ----------------------------------------------------------------------------------------------

    @Override
    public ResourceConsumerGateway getGateway() {
        return getSelfGateway(ResourceConsumerGateway.class);
    }
    
    // ----------------------------------------------------------------------------------------------
    // 组件工厂
    // ----------------------------------------------------------------------------------------------

    private static ExecutionSlotAllocatorFactory createExecutionSlotAllocatorFactory(
            Configuration configuration, SlotPool slotPool) {
        final SlotSelectionStrategy slotSelectionStrategy =
                SlotSelectionStrategyUtils.selectSlotSelectionStrategy(configuration);
        final PhysicalSlotProvider physicalSlotProvider =
                new PhysicalSlotProviderImpl(slotSelectionStrategy, slotPool);

        return new SimpleExecutionSlotAllocator.Factory(physicalSlotProvider, false);
    }
    
    // ----------------------------------------------------------------------------------------------
    // Utility classes
    // ----------------------------------------------------------------------------------------------

    // 心跳 Sender
    private static final class ResourceProviderHeartbeatSender
            			extends HeartbeatSender<AllocatedSlotReport> {
        private final ResourceProviderGateway resourceProviderGateway;

        private ResourceProviderHeartbeatSender(ResourceProviderGateway resourceProviderGateway) {
            this.resourceProviderGateway = resourceProviderGateway;
        }

        @Override
        public CompletableFuture<Void> requestHeartbeat(
                ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
            return resourceProviderGateway.heartbeatFromResourceConsumer(resourceID, allocatedSlotReport);
        }
    }

    private static final class ResourceManagerHeartbeatReceiver extends HeartbeatReceiver<Void> {
        private final ResourceManagerGateway resourceManagerGateway;

        private ResourceManagerHeartbeatReceiver(ResourceManagerGateway resourceManagerGateway) {
            this.resourceManagerGateway = resourceManagerGateway;
        }

        @Override
        public CompletableFuture<Void> receiveHeartbeat(ResourceID resourceID, Void payload) {
            return resourceManagerGateway.heartbeatFromJobManager(resourceID);
        }
    }

    // 资源管理器主节点监听
    private class ResourceManagerLeaderListener implements LeaderRetrievalListener {

        @Override
        public void notifyLeaderAddress(final String leaderAddress, final UUID leaderSessionID) {
            runAsync(
                    () ->
                            notifyOfNewResourceManagerLeader(
                                    leaderAddress,
                                    ResourceManagerId.fromUuidOrNull(leaderSessionID)));
        }

        @Override
        public void handleError(final Exception exception) {
            handleJobMasterError(
                    new Exception("Fatal error in the ResourceManager leader service", exception));
        }
    }

    // ---------------------------------------资源管理器连接-------------------------------------------------------
    private class ResourceManagerConnection
            extends RegisteredRpcConnection<
                    ResourceManagerId,
                    ResourceManagerGateway,
                    ResourceConsumerRegistrationSuccess,
                    RegistrationResponse.Rejection> {
        private final ConsumerID jobID;

        private final ResourceID consumerManagerResourceID;

        private final String consumerManagerRpcAddress;

        private final ConsumingMasterId consumingMasterId;

        ResourceManagerConnection(
                final Logger log,
                final ConsumerID jobID,
                final ResourceID jobManagerResourceID,
                final String jobManagerRpcAddress,
                final ConsumingMasterId consumingMasterId,
                final String resourceManagerAddress,
                final ResourceManagerId resourceManagerId,
                final Executor executor) {
            super(log, resourceManagerAddress, resourceManagerId, executor);
            this.jobID = checkNotNull(jobID);
            this.consumerManagerResourceID = checkNotNull(jobManagerResourceID);
            this.consumerManagerRpcAddress = checkNotNull(jobManagerRpcAddress);
            this.consumingMasterId = checkNotNull(consumingMasterId);
        }

        @Override
        protected RetryingRegistration<
                        ResourceManagerId,
                        ResourceManagerGateway,
                        ResourceConsumerRegistrationSuccess,
                        RegistrationResponse.Rejection>
                generateRegistration() {
            return new RetryingRegistration<
                    ResourceManagerId,
                    ResourceManagerGateway,
                    ResourceConsumerRegistrationSuccess,
                    RegistrationResponse.Rejection>(
                    log,
                    getRpcService(),
                    "ResourceManager",
                    ResourceManagerGateway.class,
                    getTargetAddress(),
                    getTargetLeaderId(),
                    resourceConsumerConfiguration.getRetryingRegistrationConfiguration()) {

                @Override
                protected CompletableFuture<RegistrationResponse> invokeRegistration(
                        ResourceManagerGateway gateway,
                        ResourceManagerId fencingToken,
                        long timeoutMillis) {
                    Time timeout = Time.milliseconds(timeoutMillis);
                    // 注册到资源管理
                    return gateway.registerResourceConsumer(
                    		consumingMasterId,
                            consumerManagerResourceID,
                            consumerManagerRpcAddress,
                            jobID,
                            timeout);
                }
            };
        }

        @Override
        protected void onRegistrationSuccess(final ResourceConsumerRegistrationSuccess success) {
            runAsync(
                    () -> {
                        // filter out outdated connections
                        // noinspection ObjectEquality
                        if (this == resourceManagerConnection) {
                            establishResourceManagerConnection(success);
                        }
                    });
        }

        @Override
        protected void onRegistrationRejection(RegistrationResponse.Rejection rejection) {
            handleJobMasterError(
                    new IllegalStateException(
                            "The ResourceManager should never reject a JobMaster registration."));
        }

        @Override
        protected void onRegistrationFailure(final Throwable failure) {
            handleJobMasterError(failure);
        }
    }

    // 资源提供者心跳监听
    private class ResourceProviderHeartbeatListener
            implements HeartbeatListener<ResourceProviderToResourceConsumerHeartbeatPayload, AllocatedSlotReport> {

        @Override
        public void notifyHeartbeatTimeout(ResourceID resourceID) {
            final String message =
                    String.format(
                            "Heartbeat of TaskManager with id %s timed out.",
                            resourceID.getStringWithMetadata());

            log.info(message);
            handleTaskManagerConnectionLoss(resourceID, new TimeoutException(message));
        }

        // 连接丢失处理
        private void handleTaskManagerConnectionLoss(ResourceID resourceID, Exception cause) {
            validateRunsInMainThread();
            disconnectTaskManager(resourceID, cause);
        }

        @Override
        public void notifyTargetUnreachable(ResourceID resourceID) {
            final String message =
                    String.format(
                            "TaskManager with id %s is no longer reachable.",
                            resourceID.getStringWithMetadata());

            log.info(message);
            handleTaskManagerConnectionLoss(resourceID, new ResoruceConsumerException(message));
        }

        @Override
        public void reportPayload(
                ResourceID resourceID, ResourceProviderToResourceConsumerHeartbeatPayload payload) {
            validateRunsInMainThread();
            executionDeploymentReconciler.reconcileExecutionDeployments(
                    resourceID,
                    payload.getExecutionDeploymentReport(),
                    executionDeploymentTracker.getExecutionsOn(resourceID));
        }

        @Override
        public AllocatedSlotReport retrievePayload(ResourceID resourceID) {
            validateRunsInMainThread();
            return slotPoolService.createAllocatedSlotReport(resourceID);
        }
    }

    // 资源管理器心跳监听器
    private class ResourceManagerHeartbeatListener implements HeartbeatListener<Void, Void> {

        @Override
        public void notifyHeartbeatTimeout(final ResourceID resourceId) {
            final String message =
                    String.format(
                            "The heartbeat of ResourceManager with id %s timed out.",
                            resourceId.getStringWithMetadata());
            log.info(message);

            handleResourceManagerConnectionLoss(resourceId, new TimeoutException(message));
        }

        private void handleResourceManagerConnectionLoss(ResourceID resourceId, Exception cause) {
            validateRunsInMainThread();
            if (establishedResourceManagerConnection != null
                    && establishedResourceManagerConnection
                            .getResourceManagerResourceID()
                            .equals(resourceId)) {
                reconnectToResourceManager(cause);
            }
        }

        @Override
        public void notifyTargetUnreachable(ResourceID resourceID) {
            final String message =
                    String.format(
                            "ResourceManager with id %s is no longer reachable.",
                            resourceID.getStringWithMetadata());
            log.info(message);

            handleResourceManagerConnectionLoss(resourceID, new ResoruceConsumerException(message));
        }

        @Override
        public void reportPayload(ResourceID resourceID, Void payload) {
            // nothing to do since the payload is of type Void
        }

        @Override
        public Void retrievePayload(ResourceID resourceID) {
            return null;
        }
    }

	@Override
	public CompletableFuture<ConsumingStatus> requestConsumingStatus(Time timeout) {
		return CompletableFuture.completedFuture(state);
	}
	
    @Override
    public CompletableFuture<ThreadDumpInfo> requestThreadDump(Time timeout) {
        int stackTraceMaxDepth = this.resourceConsumerConfiguration.getConfiguration().get(ClusterOptions.THREAD_DUMP_STACKTRACE_MAX_DEPTH);
        return CompletableFuture.completedFuture(ThreadDumpInfo.dumpAndCreate(stackTraceMaxDepth));
    }

    // 
    public boolean transitionState(ConsumingStatus current, ConsumingStatus newState) {
        return transitionState(current, newState, null);
    }

    private void transitionState(ConsumingStatus newState, Throwable error) {
        transitionState(state, newState, error);
    }

    private boolean transitionState(ConsumingStatus current, ConsumingStatus newState, Throwable error) {
        // consistency check
        if (current.isTerminalState()) {
            String message = "trying to leave terminal state " + current;
            LOG.error(message);
            throw new IllegalStateException(message);
        }

        // now do the actual state transition
        if (state == current) {
            state = newState;
            LOG.info(
                    "State switched from {} to {}.",
                    current,
                    newState,
                    error);
            return true;
        } else {
            return false;
        }
    }
    
}
